package com.swak.reactivex.transport.funcs;

import java.util.Set;
import java.util.concurrent.CompletionStage;

import com.swak.exception.RemotingException;
import com.swak.reactivex.transport.Channel;
import com.swak.reactivex.transport.ExchangeChannel;
import com.swak.reactivex.transport.ExchangeHandler;
import com.swak.utils.Sets;

public class ExchangeHandlers implements ExchangeHandler {

	protected Set<ChannelSentHandler> sentHandlers = null;
	protected Set<ChannelReceivedHandler> receivedHandlers = null;
	protected Set<ChannelDisConnectedHandler> disConnectedHandlers = null;
	protected Set<ChannelConnectedHandler> connectedHandlers = null;
	protected Set<ChannelCaughtHandler> caughtHandlers = null;
	protected Set<ExchangeReplyHandler> replyHandlers = null;

	public void addExchangeReplyHandler(ExchangeReplyHandler handler) {
		if (replyHandlers == null) {
			synchronized (this) {
				if (replyHandlers == null) {
					replyHandlers = Sets.newConcurrentSet();
				}
			}
		}
		replyHandlers.add(handler);
	}

	public void removeExchangeReplyHandler(ExchangeReplyHandler handler) {
		if (replyHandlers == null) {
			return;
		}
		replyHandlers.remove(handler);
	}

	public void addChannelSentHandler(ChannelSentHandler handler) {
		if (sentHandlers == null) {
			synchronized (this) {
				if (sentHandlers == null) {
					sentHandlers = Sets.newConcurrentSet();
				}
			}
		}
		sentHandlers.add(handler);
	}

	public void removeChannelSentHandler(ChannelSentHandler handler) {
		if (sentHandlers == null) {
			return;
		}
		sentHandlers.remove(handler);
	}

	public void addChannelReceivedHandler(ChannelReceivedHandler handler) {
		if (receivedHandlers == null) {
			synchronized (this) {
				if (receivedHandlers == null) {
					receivedHandlers = Sets.newConcurrentSet();
				}
			}
		}
		receivedHandlers.add(handler);
	}

	public void removeChannelReceivedHandler(ChannelReceivedHandler handler) {
		if (receivedHandlers == null) {
			return;
		}
		receivedHandlers.remove(handler);
	}

	public void addChannelDisConnectedHandler(ChannelDisConnectedHandler handler) {
		if (disConnectedHandlers == null) {
			synchronized (this) {
				if (disConnectedHandlers == null) {
					disConnectedHandlers = Sets.newConcurrentSet();
				}
			}
		}
		disConnectedHandlers.add(handler);
	}

	public void removeChannelDisConnectedHandler(ChannelDisConnectedHandler handler) {
		if (disConnectedHandlers == null) {
			return;
		}
		disConnectedHandlers.remove(handler);
	}

	public void addChannelConnectedHandler(ChannelConnectedHandler handler) {
		if (connectedHandlers == null) {
			synchronized (this) {
				if (connectedHandlers == null) {
					connectedHandlers = Sets.newConcurrentSet();
				}
			}
		}
		connectedHandlers.add(handler);
	}

	public void removeChannelConnectedHandler(ChannelConnectedHandler handler) {
		if (connectedHandlers == null) {
			return;
		}
		connectedHandlers.remove(handler);
	}

	public void addChannelCaughtHandler(ChannelCaughtHandler handler) {
		if (caughtHandlers == null) {
			synchronized (this) {
				if (caughtHandlers == null) {
					caughtHandlers = Sets.newConcurrentSet();
				}
			}
		}
		caughtHandlers.add(handler);
	}

	public void removeChannelCaughtHandler(ChannelCaughtHandler handler) {
		if (caughtHandlers == null) {
			return;
		}
		caughtHandlers.remove(handler);
	}

	@Override
	public void connected(Channel channel) throws RemotingException {
		if (connectedHandlers == null) {
			return;
		}
		for (ChannelConnectedHandler handler : connectedHandlers) {
			handler.connected(channel);
		}
	}

	@Override
	public void disconnected(Channel channel) throws RemotingException {
		if (disConnectedHandlers == null) {
			return;
		}
		for (ChannelDisConnectedHandler handler : disConnectedHandlers) {
			handler.disconnected(channel);
		}
	}

	@Override
	public void sent(Channel channel, Object message) throws RemotingException {
		if (sentHandlers == null) {
			return;
		}
		for (ChannelSentHandler handler : sentHandlers) {
			handler.sent(channel, message);
		}
	}

	@Override
	public void received(Channel channel, Object message) throws RemotingException {
		if (receivedHandlers == null) {
			return;
		}
		for (ChannelReceivedHandler handler : receivedHandlers) {
			handler.received(channel, message);
		}
	}

	@Override
	public void caught(Channel channel, Throwable exception) throws RemotingException {
		if (caughtHandlers == null) {
			return;
		}
		for (ChannelCaughtHandler handler : caughtHandlers) {
			handler.caught(channel, exception);
		}
	}

	@Override
	public CompletionStage<Object> reply(ExchangeChannel channel, Object message) throws RemotingException {
		if (replyHandlers == null) {
			return null;
		}
		for (ExchangeReplyHandler handler : replyHandlers) {
			return handler.reply(channel, message);
		}
		return null;
	}
}
